package bsp.scan;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
import org.apache.hama.bsp.BSPJob;
import org.apache.hama.bsp.BSPJobClient;
import org.apache.hama.bsp.ClusterStatus;

import util.Tool;
import util.functors.Multiply;

public class TestBSPScan {

	public static final Log LOG = LogFactory.getLog(TestBSPScan.class);

	public static class MyScan extends Scan {
		public void setOperator() {
			this.oplus = new Multiply();
			// this.oplus = new SumLong();
		}

	}

	public static void main(String[] args) throws InterruptedException,
			IOException, ClassNotFoundException {

		// BSP job configuration

		System.out.println("args.length " + args.length);
		if (args.length >= 2) {
			Path INPUT = new Path(args[0]);
			Path OUTPUT = new Path(args[1]);
			BSPJob scan = Scan.createScanJob(INPUT, OUTPUT);

			// Set the job name
			scan.setJobName("BSP Scan");
			scan.setJarByClass(TestBSPScan.class);
			scan.setBspClass(MyScan.class);

			BSPJobClient jobClient = new BSPJobClient(scan.getConfiguration());
			ClusterStatus cluster = jobClient.getClusterStatus(true);
			int outputPrintflg = -1;
			if (args.length > 2)
				scan.setNumBspTask(Integer.parseInt(args[2]));
			else
				scan.setNumBspTask(cluster.getMaxTasks());
			if (args.length > 3 && args[3].equalsIgnoreCase("debug")) {
				outputPrintflg = 1;
			}

			System.out.println("BSP tasks(bsp.setNumBspTask) = "
					+ scan.getNumBspTask());
			long startTime = System.currentTimeMillis();
			if (scan.waitForCompletion(true)) {
				// for test
				if (outputPrintflg > 0)
					Tool.printOutput(scan.getConfiguration(), OUTPUT);
				System.out.println("Job Finished in "
						+ (System.currentTimeMillis() - startTime) / 1000.0
						+ " seconds");
			}
		} else {
			System.err.println("className: INPUT OUTPUT BspTaskNum");
			System.exit(-1);
		}

		System.exit(0);
	}


}
